package cn.doitedu.data_sync;

import cn.doitedu.udfs.GeoHashUDF;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/6
 * @Desc: 学大数据，上多易教育
 *
 **/
public class Job02_GpsReference_Hbase {
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/flink/ck");

        // 本任务是一个批计算任务，用批模式跑
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射mysql中的gps参考点地理位置信息表
        tenv.executeSql(
                " create table gps_reference_mysql(                  "+
                        "     lat double,                                    "+
                        "     lng double,                                    "+
                        "     region string,                                 "+
                        "     city string,                                   "+
                        "     province string                                "+
                        " ) with (                                           "+
                        "  'connector' = 'jdbc',                             "+
                        "  'url' = 'jdbc:mysql://doitedu:3306/realtimedw',   "+
                        "  'table-name' = 'gps_tmp',                         "+
                        "  'username' = 'root',                              "+
                        "  'password' = 'root'                               "+
                        " )                                                  "
        );


        tenv.executeSql(
                " create table dim_geo_info_hbase (                  "+
                        "   geohash string                        "+
                        "   ,f ROW<                               "+
                        "   province string                       "+
                        "   ,city string                          "+
                        "   ,region string   >                    "+
                        " ) WITH (                                "+
                        "  'connector' = 'hbase-2.2',             "+
                        "  'table-name' = 'dim_gps_info',         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'    "+
                        " )                                        "
        );


        // sql：把源表的gps坐标转成 geohash
        // 考虑热点：用geohash码作为rowKey是否有热点？
        // rowKey有规律，但不见得有热点： 规律是 位置相近的数据连续存放在一起
        // 如果要到这个规律： 把geohash反转即可
        tenv.createTemporaryFunction("gps_geo", GeoHashUDF.class);

        tenv.executeSql(
                " insert into dim_geo_info_hbase  "+
                        " select                          "+
                        "                                 "+
                        " reverse(gps_geo(lat,lng)) as geohash,    "+  // 反转geohash打破相近位置在一起的规律
                        " row(province,city,region) as f  "+
                        "                                 "+
                        " from gps_reference_mysql        "+
                        " where lat is not null and lng is not null       "+
                        " and province is not null and city is not null   "+
                        " and region is not null                          "
        );


    }


}
